import cgi
import functools
import http.server
import json
import math
import socket
import time
import uuid
import socketserver
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib import parse
from urllib.parse import urlparse

import redis.exceptions
from redis.client import Redis

mconn = Redis(host="192.168.35.110", port=6379, db=0, decode_responses=True)


def acquire_lock(conn, lockname, acquire_timeout=10):
    identifier = str(uuid.uuid4())
    end = time.time() + acquire_timeout
    while time.time() < end:
        if conn.setnx('lock:' + lockname, identifier):
            return identifier
        time.sleep(0.001)
    return False


def acquire_lock_with_timeout(conn, lockname, acquire_timeout=10, lock_timeout=10):
    identifier = str(uuid.uuid4())
    lockname = 'lock:' + lockname
    lock_timeout = int(math.ceil(lock_timeout))
    end = time.time() + acquire_timeout
    while time.time() < end:
        if conn.setnx(lockname, identifier):
            conn.expire(lockname, lock_timeout)
            return identifier
        elif not conn.ttl(lockname):
            conn.expire(lockname, lock_timeout)
        time.sleep(0.001)
    return False


def release_lock(conn, lockname, identifier):
    pipe = conn.pipeline(True)
    lockname = 'lock:' + lockname
    while True:
        try:
            pipe.watch(lockname)
            if pipe.get(lockname) == identifier:
                pipe.multi()
                pipe.delete(lockname)
                pipe.execute()
            pipe.unwatch()
            break
        except redis.exceptions.WatchError:
            pass
    return False


def create_user(conn, login, name):
    llogin = login.lower()
    lock = acquire_lock_with_timeout(conn, 'user:' + llogin, 1)
    if not lock:
        return None
    if conn.hget('users:', llogin):
        release_lock(conn, 'user:' + llogin, lock)
        return None
    id = conn.incr('user:id:')
    pipe = conn.pipeline(True)
    pipe.hset('users:', llogin, id)
    pipe.hmset('user:%s' % id, {
        'login': login,
        'id': id,
        'name': name,
        'following': 0,
        'posts': 0,
        'signup': time.time(),
    })
    pipe.execute()
    release_lock(conn, 'user:' + llogin, lock)
    return id


def create_status(conn, uid, message, **data):
    pipe = conn.pipeline(True)
    pipe.hget('user:%s' % uid, 'login')
    pipe.incr('status:id:')
    login, id = pipe.execute()
    if not login:
        return None
    data.update({
        'message': message,
        'posted': time.time(),
        'id': id,
        'uid': uid,
        'login': login
    })
    pipe.hmset('status:%s' % id, data)
    pipe.hincrby('user:%s' % uid, 'posts')
    pipe.publish('streaming:status:', json.dumps(data))
    pipe.execute()
    return id


def get_status_message(conn, uid, timeline='home:', page=1, count=30):
    statuses = conn.zrevrange('%s%s' % (timeline, uid), (page - 1) * count, page * count - 1)
    pipe = conn.pipeline(True)
    for id in statuses:
        pipe.hgetall('status:%s' % id)
    reslut = pipe.execute()
    return filter(None, reslut)


HOME_TIMELINE_SIZE = 1000


def follow_user(conn, uid, other_id):
    fkey1 = 'following:%s' % uid
    fkey2 = 'followers:%s' % other_id
    if conn.zscore(fkey1, other_id):
        return None
    now = time.time()
    pipe = conn.pipeline(True)
    pipe.zadd(fkey1, {other_id: now})
    pipe.zadd(fkey2, {uid: now})
    pipe.zrevrange('profile:%s' % other_id, 0, HOME_TIMELINE_SIZE - 1, withscores=True)
    result = pipe.execute()
    following, followers, status_and_score = result[-3:]
    pipe.hincrby('user:%s' % uid, 'following', int(following))
    pipe.hincrby('user:%s' % other_id, 'followers', int(followers))
    if status_and_score:
        r1 = dict(status_and_score)
        pipe.zadd('home:%s' % uid, r1)
    pipe.zremrangebyrank('home:%s' % uid, 0, -HOME_TIMELINE_SIZE - 1)
    pipe.execute()
    return True


def unfollow_user(conn, uid, other_id):
    fkey1 = 'following:%s' % uid
    fkey2 = 'followers:%s' % other_id

    if not conn.zscore(fkey1, other_id):
        return None
    pipe = conn.pipeline(True)
    pipe.zrem(fkey1, other_id)
    pipe.zrem(fkey2, uid)
    pipe.zrevrange('profile:%s' % other_id, 0, HOME_TIMELINE_SIZE - 1)
    result = pipe.execute()
    following, followers, status = result[-3:]
    pipe.hincrby('user:%s' % uid, 'following', -int(following))
    pipe.hincrby('user:%s' % other_id, 'followers', -int(followers))
    if status:
        pipe.zrem('home:%s' % uid, *status)
    pipe.execute()
    return True


def post_status(conn, uid, message, **data):
    id = create_status(conn, uid, message, **data)
    if not id:
        return None
    posted = conn.hget('status:%s' % id, 'posted')
    if not posted:
        return None
    post = {str(id): float(posted)}
    conn.zadd('profile:%s' % uid, post)
    syndicate_status(conn, uid, post)
    return id


POSTS_PRE_PASS = 1000


def syndicate_status(conn, uid, post, start=0):
    followers = conn.zrangebyscore('followers:%s' % uid, start, 'inf', start=0, num=POSTS_PRE_PASS, withscores=True)
    pipe = conn.pipeline(False)
    for follower, start in followers:
        pipe.zadd('home:%s' % follower, post)
        pipe.zremrangebyrank('home:%s' % follower, 0, -HOME_TIMELINE_SIZE - 1)
    pipe.execute()
    if len(followers) >= POSTS_PRE_PASS:
        pass


def delete_status(conn, uid, status_id):
    key = 'status:%s' % status_id
    lock = acquire_lock_with_timeout(conn, key, 1)
    if not lock:
        return None
    if conn.hget(key, 'uid') != str(uid):
        release_lock(conn, key, lock)
        return None
    pipe = conn.pipeline(True)
    status = conn.hgetall(key)
    status['deleted'] = True
    pipe.publish('streaming:status:', json.dumps(status))
    pipe.delete(key)
    pipe.zrem('profile:%s' % uid, status_id)
    pipe.hincrby('user:%s' % uid, 'posts', -1)
    pipe.execute()
    release_lock(conn, key, lock)
    return True


def parse_identifier(handler):
    handler.identifier = None
    handler.query = {}
    if '?' in handler.path:
        url = urlparse(handler.path)
        handler.query = parse.parse_qs(url.query)
        handler.path = url.path
        identifier = handler.query.get('identifier') or [None]
        handler.identifier = identifier[0]
    return handler


FILTERS = ['track', 'filter', 'location']


def process_filters(handler):
    id = handler.identifier
    if not id:
        return handler.send_error(401, "identifier missing")
    method = handler.path.rsplit('/')[-1].split('.')[0]
    name = None
    args = None
    if method == 'filter':
        data = cgi.FieldStorage(
            fp=handler.rfile,
            headers=handler.headers,
            environ={'REQUEST_METHOD': 'POST', 'CONTENT_TYPE': handler.headers['Content-Type'],
                     })
        for name in data:
            if name in FILTERS:
                args = data.getfirst(name).lower().split('.')
                break
        if not args:
            return handler.send_error(401, "no filter provided")
    else:
        args = handler.query
    handler.send_response(200)
    handler.send_header('Transfer-Encoding', 'chunked')
    handler.end_headers()

    quit = [False]
    for item in filter_content(mconn, id, method, name, args, quit):
        try:
            item = json.dumps(item)
            handler.wfile.write(('%X\r\n%s\r\n' % (len(item), item)).encode())
        except socket.error:
            quit[0] = True
    if not quit[0]:
        handler.wfile.write(('0\r\n\r\n').encode())


def filter_content(conn, id, method, name, args, quit):
    pubsub = conn.pubsub()
    pubsub.subscribe(['streaming:status:'])
    for item in pubsub.listen():
        msg = item
        yield msg
        if quit[0]:
            break
    pubsub.reset()


class StreamingAPIServer(socketserver.ThreadingMixIn,
                         http.server.HTTPServer):
    daemon_threads = True


class StreamingAPIRequestHandler(http.server.BaseHTTPRequestHandler):

    def do_GET(self):
        parse_identifier(self)
        if self.path == '/statuses/sample.json':
            process_filters(self)
        elif self.path == '/statuses/create':
            create_status(mconn, 1, 'Hello World ' + str(time.time()))
        return self.send_error(404)

    def do_POST(self):
        parse_identifier(self)
        if self.path != '/statuses/filter.json':
            return self.send_error(404)
        process_filters(self)


CONFIGS = {}
CHECKED = {}


def get_config(conn, type, component, wait=1):
    key = 'config:%s:%s' % (type, component)
    if CHECKED.get(key) is None:
        CHECKED.setdefault(key, 0)
    if CHECKED.get(key) < time.time() - wait:
        CHECKED[key] = time.time()
        config = json.loads(conn.get(key) or '{}')
        old_config = CONFIGS.get(key)
        if config != old_config:
            CONFIGS[key] = config
    return CONFIGS.get(key)


REDIS_CONNECTIONS = {}


def redis_connection(component, wait=1):
    key = 'config:redis:' + component

    def wrapper(function):
        @functools.wraps(function)
        def call(*args, **kwargs):
            old_config = CONFIGS.get(key, object())
            _config = get_config(mconn, 'redis', component, wait)
            config = {}
            for k, v in _config.items():
                config[k] = v
            if config != old_config:
                REDIS_CONNECTIONS[key] = redis.Redis(**config)
            return function(REDIS_CONNECTIONS.get(key), *args, **kwargs)

        return call

    return wrapper


def print_hi(name):
    print(f'Hi, {name}')  # 按 Ctrl+F8 切换断点。


# 按间距中的绿色按钮以运行脚本。
if __name__ == '__main__':
    # create_user(mconn, 'HymanTS', 'HymanLiu')
    # create_user(mconn, 'TomCat', 'Tom')
    # create_status(mconn, 1, 'Hello World ' + str(time.time()))
    # result = get_status_message(mconn, 1)
    # print(list(result))
    # follow_user(mconn, 2, 1)
    # unfollow_user(mconn, 2, 1)
    # id = post_status(mconn, 1, 'Hello World ' + str(time.time()))
    # delete_status(mconn, 1, id)
    server = StreamingAPIServer(('localhost', 8080), StreamingAPIRequestHandler)
    print('Starting server, user<CTRL-C> to stop')
    server.serve_forever()

# 访问 https://www.jetbrains.com/help/pycharm/ 获取 PyCharm 帮助
